package cn.org.queue.send;

import cn.org.queue.manage.RabbitQueueManager;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

/**
 * @Package: cn.org.queue
 * @ClassName: Producter
 * @Description:
 * @Author: mac-pro
 * @CreateDate: 2019/2/14 上午11:08
 * @Version: 1.0
 */
@Slf4j
@NoArgsConstructor
public class RabbitProducerMessage extends RabbitQueueManager {
    /**
     * 消息立即发送
     *
     * @param queueName  队列名称
     * @param routingKey 路由名称
     * @param msgTplId   消息模板ID
     * @param system     系统
     * @param payload    参数内容
     */
    public void sendMessage(String queueName, String routingKey, String msgTplId, String system, Object payload) {
        byte[] bytes = JSONObject.toJSONBytes(payload);
        publish(bytes, queueName == null ? "master_queue" : queueName, routingKey == null ? "master_routing_key" : routingKey);
        log.info("即时消息发送成功 exchangeName={},routingKey={},queueName={},msgTplId={},system={},payload={}",
                EXCHANGE_NAME, routingKey == null ? "master_routing_key" : routingKey, queueName == null ? "master_queue" : queueName, msgTplId, system, new String(bytes));
    }

    /**
     * 消息延迟发送
     *
     * @param queueName  队列名称
     * @param routingKey 路由名称
     * @param msgTplId   消息模板ID
     * @param system     系统
     * @param payload    参数内容
     * @param trigger    触发时间
     */
    public void sendMessageDelay(String queueName, String routingKey, String msgTplId, String system, Map<Object, Object> payload, long trigger) {
        byte[] bytes = JSONObject.toJSONBytes(payload);
        publishDelay(bytes, queueName == null ? "master_queue" : queueName, routingKey == null ? "master_routing_key" : routingKey, trigger);
        log.info("延时消息发送成功 exchangeName={},routingKey={},queueName={},msgTplId={},system={},payload={},trigger={}",
                EXCHANGE_NAME, routingKey == null ? "master_routing_key" : routingKey, queueName == null ? "master_queue" : queueName, msgTplId, system, new String(bytes),trigger);
    }

    private void publish(byte[] payload, String queueName, String routingKey) {
        try {
            setQueueType(queueName, routingKey);
            /**
             * 发送消息到队列中
             * 参数1：交换机exchange名字，若为空则使用默认的exchange[]
             * 参数2：routing key - 路由地址
             * 参数3：其他的属性
             * 参数4：消息体
             * RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型，
             * 任何发往这个exchange的消息都会被路由到routing key的名字对应的队列上，如果没有对应的队列，则消息会被丢弃
             */
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, payload);
        } catch (Exception e) {
            log.error("send message is error", e);
        }
    }

    private void publishDelay(byte[] payload, String queueName, String routingKey, long trigger) {
        try {
            setQueueType(queueName, routingKey);
            Map<String,Object> headers = new HashMap<>();
            headers.put("x-delay",trigger);
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            AMQP.BasicProperties properties = builder.headers(headers).build();
            channel.basicPublish(EXCHANGE_NAME, routingKey, properties, payload);
        } catch (Exception e) {
            log.error("send message is error", e);
        }
    }
}
